/**
 * @brief 设备录像缓存
 * date:2023/12/18 
 * author:Coolio
 */
#ifndef ZMEDIASERVER_SRC_CACHE_DEVICERECORDCACHE_H
#define ZMEDIASERVER_SRC_CACHE_DEVICERECORDCACHE_H

#include "DeviceMediaBase.h"
#include "Util/List.h"
#include "Network/Buffer.h"
#include "Extension/Frame.h"
#include "Poller/MutableTimer.h"
#include "PrivateProtocol.h"
#include "Common/MediaSink.h"
#include "Common/MultiMediaSourceMuxer.h"


/// 合并写缓存模板
/// \tparam frame 帧类型
template<typename frame>
class FrameRecordCache  : public SerializeObject {
    typedef   std::vector<std::shared_ptr<frame> >  frame_list;
    public:
        
        //从缓存中读取数据的对象
        class CacheSubScriber: public SerializeObject {
            friend class FrameRecordCache;
            public:
                class CacheOrigin : public MediaSourceEventInterceptor {
                    public:
                        typedef std::shared_ptr<CacheOrigin> Ptr;
                        CacheOrigin(CacheSubScriber* pSubcriber):m_pSubcriber(pSubcriber){}
                        ~CacheOrigin() = default;
                        //// MediaSourceEvent override ////
                        MediaOriginType getOriginType(MediaSource &sender) const override {
                            return MediaOriginType::memcache;
                        }
                        void onReaderChanged(MediaSource &sender, int size) override {
                            m_pSubcriber->TriggerDelayedClose();                     
                        }
                    private:
                        CacheSubScriber* m_pSubcriber;
                        
                };
                CacheSubScriber(EventPoller::Ptr poller,FrameRecordCache* pCache,const privateproto::DeviceChannel& dev_chan,time_t start_push_time,time_t end_push_time);
                ~CacheSubScriber();
                //// SerializeObject override ////
                Value DumpAsJson() override;
                int StartPushRecordStream();
                void ControlStream(const std::string command,const double scale);
                void CloseSubscriber();
                bool IsClosed();
                std::string GetNewStreamId2()const;
                int GetReaderCount() const;
                //触发延迟关闭,不过需要先确认Readers为空
                void TriggerDelayedClose();
            private:
                MultiMediaSourceMuxer::Ptr      m_muxer;//与发送流紧密相关
                int                             m_curPushPosInList;//当前发送的帧在List中的位置序号,用于暂停后恢复逻辑的处理
                uint32_t                        m_curPushPts;//当前发送的帧的pts值,发送逻辑是在固定的时间间隔长度以内的pts时间范围的帧都推送出去
                float                           m_playscalevalue;//倍速播放速度(0.5/1/2/4)
                time_t                          m_start_push_time;//该次请求的开始时间
                time_t                          m_end_push_time;//该次请求的结束时间
                FrameRecordCache*               m_pCacher;//设备缓存对象指针
                //发送数据定时器
                MutableTimer::Ptr               m_send_pkt_timer;//把发送定时器放在缓冲器中是考虑到多段录像同时请求发送
                std::shared_ptr<CacheOrigin>    m_origindesc;    
                privateproto::DeviceChannel     m_dev_chan;
                Ticker                          m_last_check_cache;
                EventPoller::Ptr                m_poller;
                Timer::Ptr                      m_close_timer;
                SubscriberStateType             m_stateType;
                int                             m_noframecounter;//无数据计数,用于超时退出
        };
    public:
        typedef std::shared_ptr<FrameRecordCache> Ptr;
        FrameRecordCache(time_t start_time,DeviceMediaBase *pCache);

        virtual ~FrameRecordCache();
       
        /**
         * @brief 合并其他的缓存片段
         * @return pts修正值
         */
        int MergeOtherFrameRecordCache(const time_t new_playstart_time,FrameRecordCache &rhs) {
            #if !defined(NDEBUG)
            TimeTicker();
            #endif
            uint32_t this_end_pts_tailored = 0;
            time_t this_end_time_tailored = 0;
            bool find_merge_frame = false;
            int this_rhs_pts_offset = 0;
            m_req_start_time = new_playstart_time;
            auto it_this = m_framecache->rbegin();
            for(; it_this != m_framecache->rend(); it_this++) {
                std::shared_ptr<frame> & pframe = *it_this;
                #ifdef ENABLE_COOLIODEBUG 
                if(pframe->configFrame()||pframe->keyFrame()) {
                #else
                if(pframe->configFrame()) {
                #endif
                    this_end_pts_tailored = pframe->pts();
                    this_end_time_tailored = m_list_start_time + (this_end_pts_tailored-m_first_pts)/m_ptstimebase_const;
                    break;
                }
            }
            TraceL << "合并目标缓存片段:" << getTimeStr("%Y-%m-%d %H:%M:%S",m_list_start_time) << "需要裁剪到:" << getTimeStr("%Y-%m-%d %H:%M:%S",this_end_time_tailored);
            auto it_rhs = rhs.m_framecache->begin();
            for(;it_rhs != rhs.m_framecache->end();){
                std::shared_ptr<frame> & pframe = *it_rhs;
                #ifdef ENABLE_COOLIODEBUG 
                if(pframe->configFrame()||pframe->keyFrame()) {
                #else
                if(!pframe->configFrame()) {
                #endif
                    it_rhs = rhs.m_framecache->erase(it_rhs);
                }else {
					it_rhs++;
                    uint32_t rhs_start_pts_tailored = pframe->pts();
                    time_t rhs_start_time_tailored = rhs.m_list_start_time + (rhs_start_pts_tailored-rhs.m_first_pts)/rhs.m_ptstimebase_const;
                    if(rhs_start_time_tailored < this_end_time_tailored) {
                        TraceL << "合并源缓存片段:" << getTimeStr("%Y-%m-%d %H:%M:%S",rhs.m_list_start_time) << "需要删除:" << getTimeStr("%Y-%m-%d %H:%M:%S",rhs_start_time_tailored);
                    }else{
                        find_merge_frame = true;
                        this_rhs_pts_offset = (int)(rhs_start_pts_tailored - this_end_pts_tailored);
                        TraceL << "找到合并的起始帧:" << getTimeStr("%Y-%m-%d %H:%M:%S",this_end_time_tailored) << " vs " << getTimeStr("%Y-%m-%d %H:%M:%S",rhs_start_time_tailored) << "pts修正offset值:" << this_rhs_pts_offset;
                        break;
                    }
                }
            }
            if(find_merge_frame) {
                it_this++;
                for(auto it = m_framecache->rbegin(); it != it_this && it != m_framecache->rend(); ) {
                    m_cachedmemsize -= (*it)->size();
                    m_framecache->erase((++it).base());
                }
                for(auto it = rhs.m_framecache->begin(); it != it_rhs && it != rhs.m_framecache->end(); ) {
                    it = rhs.m_framecache->erase(it);
                }
                for(auto it = rhs.m_framecache->begin(); it != it_rhs && it != rhs.m_framecache->end(); ) {
                    std::shared_ptr<frame> & pframe = *it_this;
                    pframe->offset_dts(this_rhs_pts_offset);
                    pframe->offset_pts(this_rhs_pts_offset);
                    m_cachedmemsize += pframe->size();
                    m_framecache->push_back(pframe);
                }
                rhs.m_framecache->clear();
                rhs.m_cachedmemsize = 0;
            }
            return this_rhs_pts_offset;
        }
        //合并缓存的简单实现
        //TODO PTS修改,重复帧删除
        FrameRecordCache& operator += (const FrameRecordCache &rhs) {
            std::vector<std::shared_ptr<frame> > & list = *rhs.m_framecache.get();
            m_framecache->insert(m_framecache->end(),list.begin(),list.end());
            rhs.clear();
            return *this;
        }
     
        void inputPacket(const std::shared_ptr<frame>& pkt,int pts_offset);
        
        void clearCache();
        int64_t GetTimeLengthMS();
        time_t GetPlayBeginTime();
        time_t GetPlayEndTime();
        /**
         * 定位帧,线程安全
         */
        int LocateFrame(time_t start_push_time,time_t end_push_time);
        /**
         * 检查是否缓存耗尽,做了时间间隔处理
         * @return true 耗尽,需要触发相关事件
         */
        bool CheckCacheExhausted(const uint32_t play_pts){
            if(play_pts >= m_last_pts) {
                return true;
            }
            return int(m_last_pts-play_pts) < (m_ptstimebase_const*30);//还剩下三十秒的数据
        }
        std::shared_ptr<frame> GetFrame(size_t frameseq);
        frame_list GetFrameByRange(size_t frameseq,float time_length_second) {
            frame_list flist;
            if(frameseq >= m_framecache->size()) {
                return flist;
            }
            size_t count = (*m_framecache.get()).size();
            uint32_t framepts = (*m_framecache.get())[frameseq]->pts();
            
            for(size_t idx = frameseq; idx < count; idx++) {
                auto pframe = (*m_framecache.get())[idx];
                if((pframe->pts() -  framepts) < (uint32_t)(time_length_second*m_ptstimebase_const)) {
                    flist.push_back(pframe);
                }else {
                    break;
                }
            }
            //InfoL << "获取起始pos:" << frameseq << ",时间范围:" << time_length_second << "以内的帧,数量:" << flist.size();
            return flist;
        }
        size_t GetFrameCount();
        int GetFps() const;
        int GetPTSTimeBase() const;
        
        //开始推录像流
        std::shared_ptr<CacheSubScriber> NewCacheSubScriber(EventPoller::Ptr poller,const privateproto::DeviceChannel& dev_chan,list<Track::Ptr> tracks,time_t start_push_time,time_t end_push_time,std::string & new_stream_id) {
            m_list_subscribers.clear();//先只支持一个同时推送流
            std::shared_ptr<CacheSubScriber> subscriber = std::make_shared<CacheSubScriber>(poller,this,dev_chan,start_push_time,end_push_time);
            new_stream_id = subscriber->GetNewStreamId2();
            
            for (auto it = tracks.begin(); it != tracks.end(); ++it) {
                //if((*it)->getTrackType()==TrackVideo)
                    subscriber->m_muxer->addTrack(*it);
            }
            subscriber->m_muxer->addTrackCompleted();
            m_list_subscribers.push_back(subscriber);
            return subscriber;
        }
        std::shared_ptr<CacheSubScriber> FindCacheSubScriber(const privateproto::DeviceChannel& dev_chan) {
            auto it = m_list_subscribers.begin();
            for(;it != m_list_subscribers.end();it++) {
                if((*it)->GetNewStreamId2().compare(dev_chan.GenStreamId2()) == 0)  {
                    return (*it);
                }
            }
            return nullptr;
        }
        void CheckCacheSubScriber() {
            auto it = m_list_subscribers.begin();
            for(;it != m_list_subscribers.end();) {
                if((*it)->IsClosed())  {
                    m_list_subscribers.erase(it++);
                }else{
                    it++;
                }
            }
        }
        void CloseCacheSubScriber(const privateproto::DeviceChannel& dev_chan) {
            auto it = m_list_subscribers.begin();
            for(;it != m_list_subscribers.end();) {
                if((*it)->GetNewStreamId2().compare(dev_chan.GenStreamId2()) == 0)  {
                    TraceL << "找到:" << dev_chan.GenStreamId2() << "并关闭";
                    (*it)->CloseSubscriber();
                    it = m_list_subscribers.erase(it);
                }else{
                    it++;
                }
            }
        }
        /**
         * @overide
         */
        virtual Value DumpAsJson();
        /**
        * 清理缓存
        * @return >= 0 -  已经清理的内存大小,保留缓存对象, < 0 - 删除对象对象，再取负值就是清理的内存大小
        */
        int64_t PurgeMemory(const int machine_memory,const int process_memory);
        int64_t FrameCacheMemorySize();
        bool IsAbnormal() const;
        
    private:
        DeviceMediaBase *                           m_pCacher;
        time_t                                      m_list_start_time;//该段缓存的起始时间(unix时间戳)
        time_t                                      m_req_start_time;//该段缓存的起始时间(unix时间戳)
        time_t                                      m_visit_time;
        uint32_t                                    m_last_pts;//缓存数据的播放时间长度(毫秒)
        uint32_t                                    m_first_pts;
        std::shared_ptr<frame_list>                 m_framecache;
        int                                         m_ptstimebase_const;//pts时间单位,按照常理来说应该是90000,但在项目上发现是以毫秒做单位
        list<std::shared_ptr<CacheSubScriber>>      m_list_subscribers;//一众的订阅者
        bool                                        m_abnormal;//遇到时间戳回环,跳变等不可控情况禁止使用
        int64_t                                     m_cachedmemsize;//缓存字节数
};

class DeviceRecordCache : public DeviceMediaBase,public MediaSinkInterface {
    public:    
        typedef std::shared_ptr<DeviceRecordCache> Ptr;
        /**
         * @param autosubscribe 是否自动添加订阅者
         */
        DeviceRecordCache(const privateproto::DeviceChannel& dev_chan, bool autosubscribe = false);
        ~DeviceRecordCache();
        std::map<time_t, FrameRecordCache<Frame>::Ptr> getCacheContainer();
    public:
        bool CheckRecordCacheMerge();
        /**
         * 播放指定时间范围以内的录像资源,如果没有命中缓存,则回调失败
         * @param dev_chan 主要是因为里面的ssrc需要变化
         */
        void play_record_resources(privateproto::DeviceChannel& dev_chan,time_t startTime,time_t endTime,std::function<void(bool ,const std::string ,const std::string )> cb);
        
        void control_cache_stream(privateproto::DeviceChannel& dev_chan, const std::string command, const double scale,std::function<void(bool controlsuccess)> cb);
        
        void close_cache_stream_subscriber(privateproto::DeviceChannel& dev_chan);
        
        
    public:
        //// SerializeObject override ////
        Value DumpAsJson() override;
        void DumpAsJsonInfo(const function<void(const Json::Value& info)> &cb);
        //// DeviceMediaBase override ////
        std::string GetDesc() override; 
        /**
        * 清理缓存
        * @return >= 0 -  已经清理的内存大小,保留缓存对象, < 0 - 删除对象对象，再取负值就是清理的内存大小
        */
        int PurgeMemory(const int machine_memory,const int process_memory);
        
        /**
         * 获取设备缓存占用的内存
         */
        int64_t DeviceCacheMemorySize();
        /**
        * 删除缓存
         */
        void CleanCache();
        
        bool isEnabled();
        
        int totalReaderCount() const;
    public:
        //// MediaSinkInterface override ////
        void inputFrame(const Frame::Ptr &frame) override;
        void addTrack(const Track::Ptr & track) override;
        void addTrackCompleted() override;
        void resetTracks() override ;
    private:
        list<Track::Ptr> m_tracks;//通道先保存起来,发送缓存流的时候再添加
        bool m_haveVideo;
        bool m_haveAudio;
        bool m_autosubscribe;
        privateproto::DeviceChannel  m_dev_chan;
        std::map<time_t, FrameRecordCache<Frame>::Ptr> m_record_cache_container;//各个时间段的录像缓存容器,各个时间段不应该出现跨天的情况
        Timer::Ptr                      m_check_merge_timer;
        time_t                          m_visit_time;
};
#endif
